package com.angel.bo.message.component.rocketmq;

import com.aliyun.openservices.ons.api.*;
import com.angel.bo.message.component.rocketmq.listener.MyOnsMqListener;
import com.angel.bo.message.component.rocketmq.listener.MyOtherOnsMqListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

/**
 * @author zhangwei
 * @date 2018/9/29 18:51
 * @desc MQ消费者配置信息
 */
@Configuration
public class ConsumerConfiguration {
    /**
     * accessKey:在阿里控制台获得
     * secretKey:在阿里控制台获得
     * topic:在阿里控制台自己创建,注:阿里的消息收费基本与topic数量关系很大,注意控制创建数量
     */
    @Value("${ali.mq.accessKey}")
    private String accessKey;
    @Value("${ali.mq.secretKey}")
    private String secretKey;
    @Value("${ali.mq.topic}")
    private String topic;
    /**
     * 监听规则
     */
    @Value("${ali.mq.tag.tag1}")
    private String tag1;
    @Value("${ali.mq.tag.tag2}")
    private String tag2;

    /**
     * 消费者id 1
     * 注:不同的监听建议使用不同的消费者id,否则会造成监听规则冲突,导致部分监听失效
     */
    private final String CID_1 = "CID_1";
    /**
     * 消费者id 2
     */
    private final String CID_2 = "CID_2";

    @Autowired
    private MyOnsMqListener myOnsMqListener;
    @Autowired
    private MyOtherOnsMqListener myOtherOnsMqListener;

    /**
     * 合奏完成-发送mimc消息推送的事件监听
     *
     * @return
     */
    @Bean
    public Consumer ensembleFinishMimcListenerConsumer() {
        return startCluster(CID_1, myOnsMqListener, tag1);
    }

    /**
     * 合奏完成-发送微信消息推送的事件监听
     *
     * @return
     */
    @Bean
    public Consumer ensembleFinishWechatListenerConsumer() {
        return startBroadCasting(CID_2, myOtherOnsMqListener, tag2);
    }


    /**
     * 开启集群订阅模式的消息监听
     *
     * @param consumerId 消费者id  ConsumerId 在阿里控制台创建消费者获取
     * @param listener   监听器,自定义实现
     * @param tags       消息发送标签
     * @return 消费者
     */
    private Consumer startCluster(String consumerId, MessageListener listener, String... tags) {
        Consumer consumer = getClusteringConsumer(consumerId);
        consumer.subscribe(topic, getTags(tags), listener);
        consumer.start();
        return consumer;
    }

    /**
     * 开启广播模式的消息监听
     *
     * @param consumerId 消费者id  ConsumerId 在阿里控制台创建消费者获取
     * @param listener   监听器,自定义实现
     * @param tags       消息发送标签
     * @return 消费者
     */
    private Consumer startBroadCasting(String consumerId, MessageListener listener, String... tags) {
        Consumer consumer = getBroadCastingConsumer(consumerId);
        consumer.subscribe(topic, getTags(tags), listener);
        consumer.start();
        return consumer;
    }

    /**
     * 获取广播模式的mq消费者
     *
     * @param consumerId
     * @return
     */
    private Consumer getBroadCastingConsumer(String consumerId) {
        Properties properties = getMqCommonProperties();
        // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.ConsumerId, consumerId);
        // 订阅方式
        properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
        return ONSFactory.createConsumer(properties);
    }

    /**
     * 获取集群订阅模式的消费者
     *
     * @param consumerId
     * @return
     */
    private Consumer getClusteringConsumer(String consumerId) {
        Properties properties = getMqCommonProperties();
        // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.ConsumerId, consumerId);
        // 订阅方式
        properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
        return ONSFactory.createConsumer(properties);
    }

    /**
     * 组装公共部分的属性
     *
     * @return
     */
    private Properties getMqCommonProperties() {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.AccessKey, accessKey);
        properties.put(PropertyKeyConst.SecretKey, secretKey);
        // 设置 TCP 接入域名（公共云生产环境）
        String onsAddr = "http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet";
        properties.put(PropertyKeyConst.ONSAddr, onsAddr);
        return properties;
    }

    /**
     * 组装tag规则
     * 多个tag时以||作为tag间的间隔
     *
     * @param tags
     * @return
     */
    private String getTags(String... tags) {
        StringBuilder tag = new StringBuilder();
        if (tags == null || tags.length == 0) {
            throw new RuntimeException("消息tag不能为空");
        } else if (tags.length != 1) {
            for (String s : tags) {
                tag.append(s).append("||");
            }
            tag = new StringBuilder(tag.substring(0, tag.lastIndexOf("||")));
        } else {
            tag = new StringBuilder(tags[0]);
        }
        return tag.toString();
    }
}
